/**
 * 
 */
package com.googlecode.hifl;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

/**
 * @author Michael Sheen
 *
 */
public class BlockRecordReader extends RecordReader<LongWritable, Text>{

	private static final Log LOG = LogFactory.getLog(BlockRecordReader.class);
	
	private CompressionCodecFactory compressionCodecFactory = null;
	private long end;
	private long pos; //璁板綍琛屽彿
	private long start;
	private BlockReader blockReader;
	private LongWritable key = null;
	private Text value = null;
	
	@Override
	public void close() throws IOException {
	}

	@Override
	public LongWritable getCurrentKey() throws IOException,
			InterruptedException {
		return null;
	}

	@Override
	public Text getCurrentValue() throws IOException, InterruptedException {
		return null;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		return 0;
	}

	@Override
	public void initialize(InputSplit inputSplit, TaskAttemptContext context)
			throws IOException, InterruptedException {
		
		FileSplit split = (FileSplit) inputSplit; //鑾峰彇鍒囧壊
		Configuration conf = context.getConfiguration();
		final Path file = split.getPath();//寰楀埌鏂囦欢璺緞
		this.compressionCodecFactory = new CompressionCodecFactory(conf);
		final CompressionCodec codec = compressionCodecFactory.getCodec(file);
		
		// 鎵撳紑鏂囦欢骞跺鎵惧垏鍓插埌寮�
	    FileSystem fs = file.getFileSystem(conf);
		FSDataInputStream fileIn = fs.open(split.getPath()); // 鎵撳紑鏂囦欢
		boolean skipFirstLine = false;
		if (codec != null) {
			blockReader = new BlockReader(codec.createInputStream(fileIn), conf);
			end = Long.MAX_VALUE;
		} else {
			if (start != 0) {
				skipFirstLine = true;
				--start;
				fileIn.seek(start);
			}
			blockReader = new BlockReader(fileIn, conf);
		}
		if (skipFirstLine) { // skip first line and re-establish "start".
			start += blockReader.readBlock(new Text(), 0, (int) Math.min((long) Integer.MAX_VALUE, end - start));
		}
		this.pos = start;
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

}
